package com.example.smartgreenhouse.mqtt;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.example.smartgreenhouse.WebSocketServer.*;
import com.example.smartgreenhouse.config.MqttConfig;
import com.example.smartgreenhouse.entity.DeviceCurrent;
import com.example.smartgreenhouse.entity.EnvironmentCurrent;
import com.example.smartgreenhouse.service.*;
import com.example.smartgreenhouse.entity.Warn;
import com.example.smartgreenhouse.service.impl.*;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;

@Slf4j
@Configuration
public class MqttSubscriber {
    @Autowired
    private MqttConfig mqttConfig;
    @Autowired
    private MqttPahoClientFactory mqttClientFactory;
    @Autowired
    private SoilMoistureServiceImpl soilMoistureService;
    @Autowired
    private SoilMoistureWebSocketServer soilMoistureWebSocketServer;
    @Autowired
    private AirHumidityService airHumidityService;
    @Autowired
    private AirTemperatureService airTemperatureService;
    @Autowired
    private AirDataWebSocketServer airDataWebSocketServer;
    @Autowired
    private LightIntensityServiceImpl lightIntensityService;
    @Autowired
    private LightIntensityWebSocketServer lightIntensityWebSocketServer;
    @Autowired
    private WarnWebSocketServer warnWebSocketServer;   //预警发布

    @Autowired
    private WarnService warnService;

    @Autowired
    private PumpStatusService pumpStatusService;

    @Autowired
    private PumpStatusWebSocketServer pumpStatusWebSocketServer;

    @Autowired
    private FanStatusService fanStatusService;

    @Autowired
    private FanStatusWebSocketServer fanStatusWebSocketServer;

    @Value("${mqtt.topic.soilmoisture}")
    private String soil_moisture_topic;  //订阅土壤湿度数据

    @Value("${mqtt.topic.airdata}")
    private String air_data_topic;   //订阅空气温湿度
    @Value("${mqtt.topic.light_intensity}")
    private String light_intensity_topic;  //订阅光照强度

    @Value("${mqtt.topic.water_pump_status}")
    private String water_pump_status_topic;  //订阅水泵状态

    @Value("${mqtt.topic.fan_status}")
    private String fan_status_topic;  //订阅水泵状态

    @Autowired
    private LightStatusService lightStatusService;

    @Autowired
    private LightStatusWebSocketServer lightStatusWebSocketServer;

    @Autowired
    private EnvironmentMetricsService environmentCurrentService;

    @Autowired
    private DeviceCurrentService deviceCurrentService;

    @Value("${mqtt.topic.light_status}")
    private String light_status_topic;

    @Value("${mqtt.topic.warn}")
    private String warn_topic;

//    @Value("${mqtt.topic.water_pump_status}")
//    private String water_pump_status;  //订阅水泵状态
//
//    @Value("${mqtt.topic.fan_status}")
//    private String fan_status;  //订阅水泵状态


//    private String water_pump_control_topic;  //发布对水泵的手动控制
//
//    private String water_pump_auto_control_topic;  //发布对水泵的自动控制

    /*
    *创建一个 MQTT 客户端，用指定的 clientId 和主题从 Broker 中订阅消息，
    * 并将接收到的消息作为字符串交给 Spring Integration 的消息流处理。
    * */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId()+LocalDateTime.now(),
                        mqttClientFactory,
                        soil_moisture_topic,
                        air_data_topic,
                        light_intensity_topic,
                        water_pump_status_topic,
                        fan_status_topic,
                        light_status_topic,
                        warn_topic.split(",")[0],warn_topic.split(",")[1],
                        warn_topic.split(",")[2],warn_topic.split(",")[3]
                        );
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        return adapter;
    }

    /**
     * 这段代码是使用 Spring Integration 框架来处理 MQTT 消息流。
     * 它创建了一个名为 mqttInFlow 的集成流，用来从 MQTT 订阅主题接收消息，并对消息进行处理。
     */
    @Bean
    public IntegrationFlow mqttInFlow() {
        return IntegrationFlows.from(mqttInbound())
                .handle(message -> {
                    //System.out.println(message);
                    //获取接收到的消息的 主题（Topic）。
                    String topic = (String) message.getHeaders().get("mqtt_receivedTopic");
                    //System.out.println(topic);
                    // 获取时间戳
                    Long timestamp = (Long) message.getHeaders().get("timestamp");
                    //获取消息的 负载（Payload），即接收到的消息内容。
                    String payload = ((String) message.getPayload())
                            // 替换非法数值
                            .replace("\"light_intensity_change_rate\":nan", "\"light_intensity_change_rate\":null")
                            .replace("\"soil_moisture_change_rate\":nan", "\"soil_moisture_change_rate\":null")
                            .replace("\"air_temperature_change_rate\":nan", "\"air_temperature_change_rate\":null")
                            .replace("\"air_humidity_change_rate\":nan", "\"air_humidity_change_rate\":null");

                    //将时间戳转化为LocalDateTime
                    Instant instant = null;
                    if (timestamp != null) {
                        instant = Instant.ofEpochMilli(timestamp);
                    }
                    LocalDateTime collectionTime = null;
                    if (instant != null) {
                        collectionTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
                    }
//                    // 打印原始消息
//                    log.info("=== 收到MQTT消息 ===");
//                    log.info("主题: {}", topic);
//                    log.info("时间: {}", collectionTime);
//                    log.info("原始数据: {}", payload);
//                    log.info("=== 消息结束 ===");
                    //使用fastjson解析payload
                    JSONObject json = JSON.parseObject(payload);
                    //收到土壤湿度数据
//                    {"soil_moisture":34,"soil_moisture_change_value":0,"soil_moisture_change_rate":0.000000}
                    if(soil_moisture_topic.equals(topic)){
                        System.out.println(json);
                        int soilMoisture = json.getIntValue("soil_moisture");
                        int soilMoistureChangeValue = json.getIntValue("soil_moisture_change_value");
                        float soilMoistureChangeRate = json.getFloatValue("soil_moisture_change_rate");
                        //数据发生改变时，把传感器的数据存入到mysql中
                        if(soilMoistureChangeValue != 0 || soilMoistureChangeRate!= 0.000000){
                            soilMoistureService.saveSoilMoisture(soilMoisture,soilMoistureChangeValue,soilMoistureChangeRate,collectionTime);

                            //当数据状态发生改变是，更新环境数据库表
                            EnvironmentCurrent entity = new EnvironmentCurrent();
                            entity.setSensorId(1L); // 需要根据实际传感器ID设置
                            entity.setGreenhouseId(1L); // 需要根据实际温室ID设置
                            entity.setMetricType("soil_moisture");
                            entity.setCurrentValue(json.getIntValue("soil_moisture"));
                            entity.setCurrentChangeValue(json.getIntValue("soil_moisture_change_value"));
                            entity.setCurrentChangeRate(json.getFloatValue("soil_moisture_change_rate"));
                            entity.setCollectTime(collectionTime);
                            environmentCurrentService.upsertValue(entity);
                        }
                        //使用Websocket将当前信息发布到指定主题让前端订阅
                        // 使用 WebSocket 将当前信息发布到指定主题
                        soilMoistureWebSocketServer.sendSoilMoistureData(soilMoisture,soilMoistureChangeValue,soilMoistureChangeRate,collectionTime);
                    }
                    //收到光照强度数据
                   // {"light_intensity":340,"light_intensity_change_value":1,"light_intensity_change_rate":0.1}
                    if(light_intensity_topic.equals(topic)){
                        System.out.println(json);
                        int lightIntensity = json.getIntValue("light_intensity");
                        int lightIntensityChangeValue = json.getIntValue("light_intensity_change_value");
                        float lightIntensityChangeRate = json.getFloatValue("light_intensity_change_rate");
                        //if(lightIntensityChangeRate == )
                        //数据发生改变时，把传感器的数据存入到mysql中
                        if(lightIntensityChangeValue!= 0 || lightIntensityChangeRate!= 0.000000){
                            lightIntensityService.saveLightIntensity(lightIntensity,lightIntensityChangeValue,lightIntensityChangeRate,collectionTime);

                            EnvironmentCurrent entity = new EnvironmentCurrent();
                            entity.setSensorId(4L); // 需要根据实际传感器ID设置
                            entity.setGreenhouseId(1L);
                            entity.setMetricType("light_intensity");
                            entity.setCurrentValue(json.getIntValue("light_intensity"));
                            entity.setCurrentChangeValue(json.getIntValue("light_intensity_change_value"));
                            entity.setCurrentChangeRate(json.getFloatValue("light_intensity_change_rate"));
                            entity.setCollectTime(collectionTime);

                            environmentCurrentService.upsertValue(entity);
                        }
                        //使用Websocket将当前信息发布到指定主题让前端订阅
                        // 使用 WebSocket 将当前信息发布到指定主题
                        lightIntensityWebSocketServer.sendLightIntensity(lightIntensity,lightIntensityChangeValue,lightIntensityChangeRate,collectionTime);
                    }
                    //收到空气温湿度数据
                    if(air_data_topic.equals(topic)){
                        System.out.println(json);
                        int airTemperature = json.getIntValue("air_temperature");
                        int temperatureChangeValue = json.getIntValue("air_temperature_change_value");
                        float temperatureChangeRate = json.getFloatValue("air_temperature_change_rate");
                        int airHumidity = json.getIntValue("air_humidity");
                        int humidityChangeValue = json.getIntValue("air_humidity_change_value");
                        float humidityChangeRate = json.getFloatValue("air_humidity_change_rate");
                        //数据发生改变时，把传感器的数据存入到mysql中
                        if(temperatureChangeValue!= 0 || temperatureChangeRate!= 0.000000){
                            airTemperatureService.saveAirData(airTemperature,temperatureChangeValue,temperatureChangeRate,collectionTime);

                            // 处理空气温度
                            EnvironmentCurrent tempEntity = new EnvironmentCurrent();
                            tempEntity.setSensorId(2L);
                            tempEntity.setGreenhouseId(1L);
                            tempEntity.setMetricType("air_temperature");
                            tempEntity.setCurrentValue(json.getIntValue("air_temperature"));
                            tempEntity.setCurrentChangeValue(json.getIntValue("air_temperature_change_value"));
                            tempEntity.setCurrentChangeRate(json.getFloatValue("air_temperature_change_rate"));
                            tempEntity.setCollectTime(collectionTime);

                            environmentCurrentService.upsertValue(tempEntity);
                        }

                        if(humidityChangeValue!= 0 || humidityChangeRate!= 0.000000){
                            airHumidityService.saveAirData(airHumidity,humidityChangeValue,humidityChangeRate,collectionTime);

                            // 处理空气湿度
                            EnvironmentCurrent humidityEntity = new EnvironmentCurrent();
                            humidityEntity.setSensorId(3L);
                            humidityEntity.setGreenhouseId(1L);
                            humidityEntity.setMetricType("air_humidity");
                            humidityEntity.setCurrentValue(json.getIntValue("air_humidity"));
                            humidityEntity.setCurrentChangeValue(json.getIntValue("air_humidity_change_value"));
                            humidityEntity.setCurrentChangeRate(json.getFloatValue("air_humidity_change_rate"));
                            humidityEntity.setCollectTime(collectionTime);

                            environmentCurrentService.upsertValue(humidityEntity);
                        }
                        //使用Websocket将当前信息发布到指定主题让前端订阅
                        // 使用 WebSocket 将当前信息发布到指定主题
                        airDataWebSocketServer.sendAirData(airTemperature,temperatureChangeValue,temperatureChangeRate,airHumidity,humidityChangeValue, humidityChangeRate,collectionTime);
                    }
                    // 处理水泵状态消息
                    // 处理水泵状态消息
                    if (water_pump_status_topic.equals(topic)) {
                        try {
                            // 检查是否发生变化
                            Integer isChanged = json.getInteger("water_pump_isChanged");
                            // 解析水泵数据
                            Integer levelObj = json.getInteger("pump_level");
                            Boolean isAutoObj = json.getBoolean("pump_isAuto");

                            // 验证必要字段
                            if (levelObj == null || isAutoObj == null || isChanged == null) {
                                log.error("水泵数据不完整:isChanged={} level={}, isAuto={}",isChanged, levelObj, isAutoObj);
                                return;
                            }

                            // 验证数据有效性
                            int level = levelObj;
                            if (level < 0 || level > 3) {
                                log.error("无效的水泵挡位数据: {}", level);
                                return;
                            }

                            // 获取或设置默认水泵ID
                            Long pumpId = json.containsKey("pump_id") ? json.getLong("pump_id") : 2L;

                            // 更新数据库和推送WebSocket
                            Long greenhouseId = 1L;
                            pumpStatusWebSocketServer.sendPumpStatus(pumpId, level, isAutoObj,greenhouseId);
                            if(isChanged == 1)
                            {
                                pumpStatusService.insertPumpStatus(pumpId, level, isAutoObj,greenhouseId);

                                DeviceCurrent entity = new DeviceCurrent();
                                entity.setDeviceType("water_pump");
                                entity.setDeviceId(pumpId);
                                entity.setCurrentLevel(levelObj);
                                entity.setIsAuto(isAutoObj);

                                deviceCurrentService.upsertDeviceStatus(entity);
                            }

                        } catch (Exception e) {
                            log.error("处理水泵状态异常", e);
                        }
                    }
                    if (fan_status_topic.equals(topic)) {
                        try {
                            // 检查是否发生变化
                            Integer isChanged = json.getInteger("fan_isChanged");
                            // 解析风扇数据
                            Integer levelObj = json.getInteger("fan_level");
                            Boolean isAutoObj = json.getBoolean("fan_isAuto");

                            // 验证必要字段
                            if (levelObj == null || isAutoObj == null || isChanged == null) {
                                log.error("风扇数据不完整:isChanged={}  level={}, isAuto={}",isChanged, levelObj, isAutoObj);
                                return;
                            }

                            // 验证数据有效性
                            int level = levelObj;
                            if (level < 0 || level > 3) {
                                log.error("无效的风扇挡位数据: {}", level);
                                return;
                            }

                            // 获取或设置默认风扇ID
                            Long fanId = json.getLong("fan_id") != null ? json.getLong("fan_id") : 1L;

                            // 更新数据库和推送WebSocket
                            fanStatusWebSocketServer.sendFanStatus(fanId, level, isAutoObj);
                            if(isChanged == 1){
                                fanStatusService.insertFanStatus(fanId, level, isAutoObj,1L);

                                DeviceCurrent entity = new DeviceCurrent();
                                entity.setDeviceType("fan");
                                entity.setDeviceId(fanId);
                                entity.setCurrentLevel(level);
                                entity.setIsAuto(isAutoObj);

                                deviceCurrentService.upsertDeviceStatus(entity);
                            }

                        } catch (Exception e) {
                            log.error("处理风扇状态异常", e);
                        }
                    }
                    if (light_status_topic.equals(topic)) {
                        try {
                            Integer isChanged = json.getInteger("light_isChanged");
                            Integer isOn = json.getInteger("light_isOn");
                            Boolean isAuto = json.getBoolean("light_isAuto");

                            if (isOn == null || isAuto == null || isChanged == null) {
                                log.error("灯光数据不完整: isChanged={} isOn={}, isAuto={}", isChanged, isOn, isAuto);
                                return;
                            }

                            Long lightId = json.getLong("light_id") != null ? json.getLong("light_id") : 3L;

                            // 推送WebSocket
                            lightStatusWebSocketServer.sendLightStatus(lightId, isOn, isAuto, 1L);
                            if(isChanged == 1) {
                                lightStatusService.insertLightStatus(lightId, isOn, isAuto, 1L);

                                DeviceCurrent entity = new DeviceCurrent();
                                entity.setDeviceType("full_spectrum_LED_light");
                                entity.setDeviceId(lightId);
                                entity.setCurrentLevel(isOn); // 需要确认字段对应关系
                                entity.setIsAuto(isAuto);

                                deviceCurrentService.upsertDeviceStatus(entity);
                            }
                        } catch (Exception e) {
                            log.error("处理灯光状态异常", e);
                        }
                    }


                    //收到预警信息,并处理预警信息
                    this.solve(topic,json,collectionTime);
                })
                .get();
    }

    /*
    * MessageChannel：Spring Integration 中的 MessageChannel 是消息传输的媒介。消息通道充当了生产者与消费者之间的桥梁。
    * DirectChannel：是 Spring Integration 提供的一种消息通道实现。它用于同步地将消息传递给消息处理器。
    * DirectChannel 是最常见的通道类型，它会直接将消息传递给下游的消息处理器，通常用于简单的应用场景。
    * @Bean：这表示 Spring 会将 mqttOutboundChannel 方法的返回值注册为一个 Bean，可以在其他地方进行注入和使用。
    * */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler(mqttConfig.getClientId(), mqttClientFactory);
        messageHandler.setAsync(true);
        ///messageHandler.setDefaultTopic(topic);
        return messageHandler;
    }
    public void solve(String topic, JSONObject json,LocalDateTime collectTime) {
        boolean isSoilMoistureWarn = warn_topic.split(",")[0].equals(topic);
        boolean isAirTemperatureWarn = warn_topic.split(",")[1].equals(topic);
        boolean isAirHumidityWarn = warn_topic.split(",")[2].equals(topic);
        boolean isSoilTemperatureWarn = warn_topic.split(",")[3].equals(topic);
        if(isSoilMoistureWarn || isAirTemperatureWarn || isAirHumidityWarn || isSoilTemperatureWarn){
            json.put("collection_time",collectTime);
            warnWebSocketServer.warn(topic,json);
            //存入数据库
            Warn warn = new Warn();
            warn.setType(json.getString("type"));
            warn.setValue(json.getIntValue("value"));
            warn.setMessage(json.getString("message"));
            warn.setCollectionTime(json.getLocalDateTime("collection_time"));
            warnService.saveWarn(warn);
        }
    }
}
